package cn.icanci.loopstack.ras.client.server;

import cn.hutool.json.JSONUtil;
import cn.icanci.loopstack.ras.client.cache.holder.RasClientRepositoryHolder;
import cn.icanci.loopstack.ras.common.socket.RasRefreshDTO;
import cn.icanci.loopstack.ras.common.socket.SocketMessage;
import cn.icanci.loopstack.ras.common.socket.UriConstant;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.*;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.CharsetUtil;
import io.netty.util.internal.ThrowableUtil;

import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author icanci
 * @since 1.0 Created in 2023/01/06 22:49
 */
@SuppressWarnings("all")
public class NamedNettyServerHandler extends SimpleChannelInboundHandler<FullHttpRequest> {

    private static final Logger              logger    = LoggerFactory.getLogger(NamedNettyServerHandler.class);

    private static RasClientRepositoryHolder rasClientRepositoryHolder;

    private static final int                 CORE_SIZE = Runtime.getRuntime().availableProcessors();

    private static final ThreadPoolExecutor  POOL      = new ThreadPoolExecutor(CORE_SIZE,                      //
        CORE_SIZE << 1,                                                                                         //
        60L,                                                                                                    //
        TimeUnit.SECONDS,                                                                                       //
        new LinkedBlockingQueue<>(2000),                                                                        //
        runnable -> new Thread(runnable, "RasServerThread Pool-" + runnable.hashCode()),                        //
        (r, executor) -> {
            throw new RuntimeException("RasServerThread Pool is EXHAUSTED!");
        });;

    public NamedNettyServerHandler() {
    }

    public static void setRasRepositoryHolder(RasClientRepositoryHolder rasClientRepositoryHolder) {
        NamedNettyServerHandler.rasClientRepositoryHolder = rasClientRepositoryHolder;
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest msg) throws Exception {
        String requestData = msg.content().toString(CharsetUtil.UTF_8);
        String uri = msg.uri();
        HttpMethod httpMethod = msg.method();
        boolean keepAlive = HttpUtil.isKeepAlive(msg);
        // 对于这种配置数据，会有很频繁的变更
        POOL.execute(() -> {
            Object responseObj = process(httpMethod, uri, requestData);

            String responseJson = JSONUtil.toJsonStr(responseObj);

            writeResponse(ctx, keepAlive, responseJson);
        });
    }

    /**
     * 后置处理
     * 
     * @param httpMethod httpMethod
     * @param uri uri
     * @param requestData requestData
     * @return Object
     */
    private Object process(HttpMethod httpMethod, String uri, String requestData) {
        if (HttpMethod.POST != httpMethod) {
            return SocketMessage.fail("Only post requests are supported");
        }
        if (StringUtils.isBlank(uri)) {
            return SocketMessage.fail("Request uri is null");
        }
        try {
            switch (uri) {
                case UriConstant.ServerToClient.HEARTBEAT:
                    logger.info("[{}][NamedNettyServerHandler][process] heartbeat", Thread.currentThread().getName());
                    return SocketMessage.success();
                case UriConstant.ServerToClient.REFRESH:
                    // 在进行刷新的时候，如果是第一次，则是全局进行加载
                    // 如果是增量数据，则是部分刷新，针对Client来说，对其来说只是刷新本地的数据，对到来的是什么数据不关心
                    // 因此可以使用同一个方法进行处理
                    rasClientRepositoryHolder.refresh(JSONUtil.toBean(requestData, RasRefreshDTO.class));
                    logger.info("[{}][NamedNettyServerHandler][process] {} was refreshed!", Thread.currentThread().getName(), requestData);
                    return SocketMessage.success();
                default:
                    return SocketMessage.fail("Invalid request, uri-mapping(" + uri + ") not found");
            }
        } catch (Throwable e) {
            logger.error("[{}][NamedNettyServerHandler][process] ex,error msg:{}", e, Thread.currentThread().getName(), ThrowableUtil.stackTraceToString(e));
            return SocketMessage.fail(ThrowableUtil.stackTraceToString(e));
        }
    }

    private void writeResponse(ChannelHandlerContext ctx, boolean keepAlive, String responseJson) {
        FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK, Unpooled.copiedBuffer(responseJson, CharsetUtil.UTF_8));
        response.headers().set(HttpHeaderNames.CONTENT_TYPE, "text/html;charset=UTF-8");
        response.headers().set(HttpHeaderNames.CONTENT_LENGTH, response.content().readableBytes());
        if (keepAlive) {
            response.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE);
        }
        ctx.writeAndFlush(response);
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.flush();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        logger.error(ThrowableUtil.stackTraceToString(cause));
        ctx.close();
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            ctx.channel().close();
        } else {
            super.userEventTriggered(ctx, evt);
        }
    }
}
